package transcode

import (
	"encoding/json"
	"time"

	"code.byted.org/bytertc/whiteboard_server/biz/pkg/mmetrics"

	"code.byted.org/bytertc/whiteboard_server/biz/dal/ddl"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/biz_error"
	"code.byted.org/bytertc/whiteboard_server/biz/pkg/mq"
	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode/callback"
	"code.byted.org/bytertc/whiteboard_server/biz/service/transcode/task"
	"code.byted.org/gopkg/ctxvalues"
	"code.byted.org/gopkg/env"
	"code.byted.org/gopkg/logs/v2/log"
	"code.byted.org/gopkg/metrics"
	"code.byted.org/videoarch/universal_utils_go/event_collector/tracer"
	"code.byted.org/videoarch/universal_utils_go/event_collector/util"
	"github.com/pkg/errors"
	"github.com/spf13/cast"
	"golang.org/x/net/context"
)

func (m *manager) handleTaskStart(ctx context.Context, eventData []byte) error {
	data := &callback.TaskStartData{}
	err := json.Unmarshal(eventData, data)
	if err != nil {
		return errors.WithStack(err)
	}
	t, err := m.repo.GetTranscodeTask(ctx, &task.TaskUnique{
		AppId:  data.GetAppId(),
		TaskId: data.GetTaskId(),
	})
	if err != nil {
		return errors.WithStack(err)
	}
	tt := &task.Task{}
	err = json.Unmarshal([]byte(t.Meta), tt)
	if err != nil {
		return errors.WithStack(err)
	}

	log.V1.CtxInfo(ctx, "callback data:%#v", data)
	err = m.repo.UpdateTranscodeTask(ctx,
		&task.TaskUnique{
			AppId:  data.GetAppId(),
			TaskId: data.GetTaskId()},
		map[string]interface{}{
			"status":   task.Running,
			"executor": data.Executor,
		})
	if err != nil {
		return errors.WithStack(err)
	}

	traceKey := &util.KeyWord{
		AppId:  tt.AppId,
		UserId: tt.UserId,
		TaskId: tt.TaskId,
	}
	otherInfo := map[string]string{
		"psm":            env.PSM(),
		"transcode_mode": cast.ToString(int(tt.TranscodeMode)),
		"input_format":   tt.TranscodeConfig.GetInputFormat(),
		"output_format":  tt.TranscodeConfig.GetOutputFormat(),
		"priority":       cast.ToString(int(tt.Priority)),
	}
	tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_task_start", "", traceKey, otherInfo)
	return nil
}

func (m *manager) handleTaskProcess(ctx context.Context, eventData []byte) error {
	data := &callback.TaskProcessData{}
	err := json.Unmarshal(eventData, data)
	if err != nil {
		return errors.WithStack(err)
	}
	t, err := m.repo.GetTranscodeTask(ctx, &task.TaskUnique{
		AppId:  data.GetAppId(),
		TaskId: data.GetTaskId(),
	})
	if err != nil {
		return errors.WithStack(err)
	}
	tt := &task.Task{}
	err = json.Unmarshal([]byte(t.Meta), tt)
	if err != nil {
		return errors.WithStack(err)
	}

	log.V1.CtxInfo(ctx, "callback data:%#v", data)
	err = m.repo.UpdateTranscodeTask(ctx,
		&task.TaskUnique{
			AppId:  data.GetAppId(),
			TaskId: data.GetTaskId()},
		map[string]interface{}{
			"status":      task.Running,
			"executor":    data.Executor,
			"file_size":   data.FileSize,
			"total_pages": data.TotalPages,
			"cur_pages":   data.CurPages,
		})

	if err != nil {
		return errors.WithStack(err)
	}

	if data.CurPages == 1 || data.CurPages == data.TotalPages || data.CurPages%10 == 0 {
		traceKey := &util.KeyWord{
			AppId:  tt.AppId,
			UserId: tt.UserId,
			TaskId: tt.TaskId,
		}
		otherInfo := map[string]string{
			"psm":            env.PSM(),
			"transcode_mode": cast.ToString(int(tt.TranscodeMode)),
			"input_format":   tt.TranscodeConfig.GetInputFormat(),
			"output_format":  tt.TranscodeConfig.GetOutputFormat(),
			"priority":       cast.ToString(int(tt.Priority)),
			"size":           cast.ToString(data.FileSize),
			"total_pages":    cast.ToString(data.TotalPages),
			"cur_pages":      cast.ToString(data.CurPages),
		}
		tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_task_process", "", traceKey, otherInfo)
	}

	return nil
}

func (m *manager) handleTaskFinish(ctx context.Context, eventData []byte) error {
	data := &callback.TaskFinishData{}
	err := json.Unmarshal(eventData, data)
	if err != nil {
		return errors.WithStack(err)
	}
	t, err := m.repo.GetTranscodeTask(ctx, &task.TaskUnique{
		AppId:  data.GetAppId(),
		TaskId: data.GetTaskId(),
	})
	if err != nil {
		return errors.WithStack(err)
	}
	tt := &task.Task{}
	err = json.Unmarshal([]byte(t.Meta), tt)
	if err != nil {
		return errors.WithStack(err)
	}

	log.V1.CtxInfo(ctx, "callback data:%#v", data)
	if data.Success {
		rs := make([]*ddl.TranscodeResult, 0)
		for _, r := range data.Result.Images {
			rs = append(rs, &ddl.TranscodeResult{
				AppId:    data.GetAppId(),
				TaskId:   data.GetTaskId(),
				FileName: data.Result.FileName,
				PageID:   r.PageID,
				ImgUrl:   r.Img,
				ThmUrl:   r.ThumbnailUrl,
			})
		}

		err = m.repo.UpdateTranscodeTask(ctx,
			&task.TaskUnique{
				AppId:  data.GetAppId(),
				TaskId: data.GetTaskId()},
			map[string]interface{}{
				"status": task.Done,
			})
		if err != nil {
			return errors.WithStack(err)
		}
		err = m.repo.BatchCreateTranscodeResult(ctx, rs)
		if err != nil {
			return errors.WithStack(err)
		}

		err = m.sendToBill(ctx, t.AppId, t.TaskId, t.TranscodeMode, t.TotalPages)

	} else {
		err = m.repo.UpdateTranscodeTask(ctx,
			&task.TaskUnique{
				AppId:  data.GetAppId(),
				TaskId: data.GetTaskId()},
			map[string]interface{}{
				"status":   task.Failed,
				"err_code": data.ErrCode,
				"err_msg":  data.ErrMsg,
			})
	}

	if err != nil {
		return errors.WithStack(err)
	}

	traceKey := &util.KeyWord{
		AppId:  tt.AppId,
		UserId: tt.UserId,
		TaskId: tt.TaskId,
	}
	result := "success"
	if !data.Success {
		result = "fail"
	}
	otherInfo := map[string]string{
		"psm":            env.PSM(),
		"transcode_mode": cast.ToString(int(tt.TranscodeMode)),
		"input_format":   tt.TranscodeConfig.GetInputFormat(),
		"output_format":  tt.TranscodeConfig.GetOutputFormat(),
		"priority":       cast.ToString(int(tt.Priority)),
		"size":           cast.ToString(t.FileSize),
		"total_pages":    cast.ToString(t.TotalPages),
		"cur_pages":      cast.ToString(t.CurPages),
		"result":         result,
		"fail_reason":    cast.ToString(data.ErrCode),
	}
	tracer.MileStone(tracer.STREAM_COLLECTOR, "transcode_finish", "", traceKey, otherInfo)
	return nil
}

func (m *manager) handleStats(ctx context.Context, eventData []byte) error {

	data := &callback.StatsData{}
	err := json.Unmarshal(eventData, data)
	if err != nil {
		return errors.WithStack(err)
	}
	log.V1.CtxInfo(ctx, "stats data:%#v", data)
	m.metricsClient.EmitStore(mmetrics.MetricsTranscodeMachineCPU, data.Cpu, metrics.T{"win_ip", data.IP})
	m.metricsClient.EmitStore(mmetrics.MetricsTranscodeMachineMem, data.Mem, metrics.T{"win_ip", data.IP})
	m.metricsClient.EmitStore(mmetrics.MetricsTranscodeMachineDisk, data.Disk, metrics.T{"win_ip", data.IP})
	m.metricsClient.EmitStore(mmetrics.MetricsTranscodeMachineNumGo, data.NumGoroutine, metrics.T{"win_ip", data.IP})
	m.metricsClient.EmitStore(mmetrics.MetricsTranscodeMachineTaskNum, data.TaskNumber, metrics.T{"win_ip", data.IP})
	return nil
}

func (m *manager) sendToBill(ctx context.Context, appID, taskID string, transcodeMode, pageNumber int) error {
	msgId := ctxvalues.LogIDDefault(ctx)
	msg := mq.DurationBillMsg{
		Type:      "whiteboard_transcode",
		MsgId:     msgId,
		Timestamp: time.Now().UnixNano() / 1e6,
		Data: mq.TranscodeData{
			AppId:         appID,
			TaskId:        taskID,
			TranscodeMode: transcodeMode,
			PageNumber:    pageNumber,
		},
	}

	data, err := json.Marshal(msg)
	if err != nil {
		return biz_error.EncodeError.Wrap(errors.Errorf("json marshal fail msg:%+v", msg))
	}
	err = m.BillSender.SendMessage(ctx, m.BillTopic, string(data))
	if err != nil {
		//todo:异常处理
		log.V1.CtxError(ctx, "send kafka fail err:%+v,data:%s", err, data)
	}
	return nil
}
